# -*- coding: utf-8 -*-
import logging
import time

# 导入必要的库
from pyspark.sql import SparkSession
from pyspark.sql.functions import *


if __name__ == '__main__':
    filename = "rent.csv"

    logging.info("开始spark分析")
    start_time = time.time()

    # 创建SparkSession
    spark = SparkSession.builder.\
        master("local").\
        appName("rent_als"). \
        config("spark.sql.shuffle.partitions", 2). \
        getOrCreate()

    # 读取CSV文件
    rent_df = spark.read.csv(filename, header=True)

    # 过滤掉贵的离谱的写字楼
    rent_df = rent_df.filter(rent_df.price >= 50).filter(rent_df.price <= 40000)

    # 租房类型分析
    print("租房类型分析:")
    rent_type_count = rent_df.groupBy("type").count().orderBy("count", ascending=False)
    rent_type_count.show()

    # 小区租房数量分析
    print("小区租房数量分析:")
    community_count = rent_df.groupBy("address").count().orderBy("count", ascending=False)
    community_count.show()

    # 小区租房均价分析
    print("小区租房均价分析:")
    community_avg_price = rent_df.groupBy("address").agg(avg("price")).orderBy("avg(price)", ascending=False)
    community_avg_price.show()

    # 租房价格范围分析
    print("租房价格范围分析:")
    rent_price_range = rent_df.groupBy("type").agg(min("price"), max("price"))
    rent_price_range.show()

    # 居室类型分析
    print("居室类型分析:")
    room_count = rent_df.groupBy("house_type").count().orderBy("count", ascending=False)
    room_count.show()

    # 地理位置分析
    print("地理位置分析:")
    area_count = rent_df.groupBy("area").count().orderBy("count", ascending=False)
    area_count.show()

    # 户型结构分析
    print("户型结构分析:")
    layout_count = rent_df.groupBy("room_orientation").count().orderBy("count", ascending=False)
    layout_count.show()

    # 朝向分析
    print("朝向分析:")
    orientation_count = rent_df.groupBy("room_orientation").count().orderBy("count", ascending=False)
    orientation_count.show()

    # 楼层分析
    print("楼层分析:")
    floor_count = rent_df.groupBy("height").count().orderBy("count", ascending=False)
    floor_count.show()

    # 房屋装修分析
    print("房屋装修分析:")
    decoration_count = rent_df.groupBy("labels").count().orderBy("count", ascending=False)
    decoration_count.show()

    # 供暖方式分析
    print("供暖方式分析:")
    heating_count = rent_df.groupBy("labels").agg(
        sum(when(col("labels").contains("集中供暖"), 1).otherwise(0)).alias("central_heating_count"),
        sum(when(~col("labels").contains("集中供暖"), 1).otherwise(0)).alias("individual_heating_count"))
    heating_count.show()

    # 看房时间分析
    print("看房时间分析:")
    watch_time_count = rent_df.groupBy("labels").agg(
        sum(when(col("labels").contains("随时看房"), 1).otherwise(0)).alias("watch_time_count"),
        sum(when(~col("labels").contains("随时看房"), 1).otherwise(0)).alias("appointment_count"))
    watch_time_count.show()

    # 标签分析
    print("标签分析:")
    label_count = rent_df.withColumn("label", split(col("labels"), ",")).select(
        explode("label").alias("label")).groupBy("label").count().orderBy("count", ascending=False)
    label_count.show()

    # 房屋面积分析
    print("房屋面积分析:")
    area_price = rent_df.select("floor_area", "price").groupBy("floor_area").agg(avg("price")).orderBy("floor_area")
    area_price.show()
